package cn.codergege.demo.batch

import cn.codergege.demo.entity.Person
import org.springframework.batch.core.Job
import org.springframework.batch.core.Step
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory
import org.springframework.batch.core.configuration.annotation.StepScope
import org.springframework.batch.core.launch.support.RunIdIncrementer
import org.springframework.batch.core.launch.support.SimpleJobLauncher
import org.springframework.batch.core.repository.JobRepository
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean
import org.springframework.batch.core.repository.support.SimpleJobRepository
import org.springframework.batch.item.ItemProcessor
import org.springframework.batch.item.ItemWriter
import org.springframework.batch.item.database.*
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder
import org.springframework.batch.item.database.support.SqlPagingQueryProviderFactoryBean
import org.springframework.batch.item.file.FlatFileItemReader
import org.springframework.batch.item.file.FlatFileItemWriter
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper
import org.springframework.batch.item.file.transform.BeanWrapperFieldExtractor
import org.springframework.batch.item.file.transform.DelimitedLineAggregator
import org.springframework.batch.support.DatabaseType
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.beans.factory.annotation.Qualifier
import org.springframework.beans.factory.annotation.Value
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.context.annotation.PropertySource
import org.springframework.core.env.Environment
import org.springframework.core.io.FileSystemResource
import org.springframework.core.task.SimpleAsyncTaskExecutor
import org.springframework.jdbc.core.BeanPropertyRowMapper
import org.springframework.transaction.PlatformTransactionManager
import org.springframework.util.ClassUtils
import javax.persistence.EntityManagerFactory
import javax.sql.DataSource

/**
 *  配置一个 batch:
 *      1. Reader
 *      2. ItemProcess
 *      3. Writer, 如果写入文件, 可能需要一个 StringHeaderWriter(已经写好)
 *      4. Step, 将上面上个组合起来
 *      5. Job, 可能需要 listener
 *      6. JobLauncher
 *          1) 如果是写入数据库, 参考 mysqlJobLauncher, 同时配备 mysqlJobRepository
 *          2) 如果是写入 Flat File, 参考 fileJobLauncher, 同时配备默认的 jobRepository
 */

@PropertySource("classpath:/batch.properties")
@Configuration
@EnableBatchProcessing
class BatchConfiguration {
    // 多数据源的时候, 这里需要考虑一下
    @Autowired
    lateinit var emf: EntityManagerFactory
    @Autowired
    lateinit var jobBuilderFactory: JobBuilderFactory
    @Autowired
    lateinit var stepBuilderFactory: StepBuilderFactory
    @Autowired
    lateinit var env: Environment

    val csvFilePath = "csv.to.mysql.person.source.path"
    val jdbcSql = "csv.to.mysql.person.jdbc.sql"
    val chunck = "csv.to.mysql.person.jdbc.chunk"
    val exportCsvHeader = "csv.to.csv.export.file.header"
    val exportCsvPath = "csv.to.csv.export.file.path"
    val pageSize = "read.from.mysql.page.size"

    // tag::Job[]
    @Bean
    fun mysql2CsvPersonJob(
        listener: JobCompletionNotificationListener,
        @Qualifier("mysql2CsvPersonStep1") step1: Step
    ): Job {
        return jobBuilderFactory
            .get("==[从 Mysql 数据库读取, 并写入 csv 文件]==")
            .incrementer(RunIdIncrementer())
            .listener(listener)
            .flow(step1)
            .end()
            .build()
    }
    @Bean
    fun csv2MysqlPersonJob(listener: JobCompletionNotificationListener
                           , @Qualifier("csv2MysqlPersonStep1") step1: Step): Job {
        // job name, 监听类可以通过 jobExecution.jobInstance.jobName 访问
        return jobBuilderFactory.get("==[导入 Person csv 文件]==")
            .incrementer(RunIdIncrementer())
            .listener(listener)
            // 只包含一个 step
            .flow(step1)
            .end()
            .build()
    }

    /**
     *  使用 fileJobLauncher
     */
    @Bean
    fun csv2CsvPersonJob(
        listener: JobCompletionNotificationListener,
        @Qualifier("csv2CsvPersonStep1") step1: Step
    ): Job {
        return jobBuilderFactory
            .get("==[从 csv 导入, 导出为 csv]==")
            .incrementer(RunIdIncrementer())
            .listener(listener)
            .flow(step1)
            .end()
            .build()
    }
    // end::Job[]


    // tag::读写处理器[]
    /**
     *  选择 普通 reader 或者 triggerReader, 需要:
     *      1. 修改后面的 step
     *      2. 计划任务中, 或控制器中的 jobLauncher.run() 方法也要修改(jobParameters 根据需要添加 addXxx)
     */
    @Bean
    @StepScope
    fun csvPersonTriggerReader(@Value("#{jobParameters['path2File']}") path2File: String): FlatFileItemReader<Person> {
        return FlatFileItemReaderBuilder<Person>()
            // 命名
            .name("csv2MysqlPersonTrigger")
            // 指定文件位置
            .resource(FileSystemResource(path2File))
            // 如果中文乱码解析异常, 加入 encoding 设置
            .encoding("GBK")
            // 指定分隔符
            .delimited().delimiter(",")
            // 字段名
            .names(arrayOf("name", "email"))
            // 忽略第一行
            .linesToSkip(1)
            // 字段映射
            .fieldSetMapper(object : BeanWrapperFieldSetMapper<Person>() {
                init {
                    setTargetType(Person::class.java)
                }
            })
            .build()
    }

    @Bean
    fun csvPersonReader(): FlatFileItemReader<Person> {
        return FlatFileItemReaderBuilder<Person>()
            // 命名
            .name("csv2MysqlPerson")
            // 指定文件位置
//            .resource(ClassPathResource("sample-data.csv"))
            .resource(FileSystemResource(env.getRequiredProperty(csvFilePath)))
            // 分隔符
            .delimited().delimiter(";")
            // 字段名
            .names(arrayOf("name", "email"))
            // 忽略第一行
            .linesToSkip(1)
            // 字段映射
            .fieldSetMapper(object : BeanWrapperFieldSetMapper<Person>() {
                init {
                    setTargetType(Person::class.java)

                }
            })
            .build()
    }

    /**
     *  使用 JdbcPagingItemReader 分页读取数据库
     */
    @Bean
    @StepScope
    fun mysqlPersonJdbcPagingItemReader(
        @Value("#{jobParameters['sqlParamName']}") name: String,
        dataSource: DataSource
    ): JdbcPagingItemReader<Person> {
        val parameterValues = HashMap<String, Any>()
        parameterValues["name"] = name

        return JdbcPagingItemReaderBuilder<Person>()
            .name(ClassUtils.getShortName(JdbcPagingItemReader::class.java))
            .dataSource(dataSource)
            .queryProvider(mysqlPersonQueryProvider(dataSource))
            .parameterValues(parameterValues)
            // 这里得自定义一个 rowMapper
            .rowMapper(PersonRowMapper())
            .pageSize(env.getRequiredProperty(pageSize).toInt())
            .build()
    }

    /**
     *  为 JdbcPagingItemReader 提供 provider
     */
    @Bean
    fun mysqlPersonQueryProvider(dataSource: DataSource): PagingQueryProvider? {
        val provider: SqlPagingQueryProviderFactoryBean = SqlPagingQueryProviderFactoryBean();
        provider.setDataSource(dataSource)
        provider.setSelectClause("select id, name, email")
        provider.setFromClause("from person")
        provider.setWhereClause("where name not like :name ")
        provider.setSortKey("id")

        return provider.`object`
    }

    @Bean
    fun personItemProcessor(): PersonItemProcessor {
        return PersonItemProcessor()
    }

    @Bean
    fun mysqlPersonJdbcWriter(dataSource: DataSource): JdbcBatchItemWriter<Person> {
        return JdbcBatchItemWriterBuilder<Person>()
            .itemSqlParameterSourceProvider(BeanPropertyItemSqlParameterSourceProvider<Person>())
//            .sql("INSERT INTO person(name, gender, birthday, email) VALUES(:name, :gender, :birthday, :email)")
            .sql(env.getRequiredProperty(jdbcSql))
            .dataSource(dataSource)
            .build()
    }

    @Bean
    fun mysqlPersonJpaWriter(): JpaItemWriter<Person> {
        val writer = JpaItemWriter<Person>()
        writer.setEntityManagerFactory(emf)
        writer.afterPropertiesSet()
        return writer
    }

    @Bean
    fun csvPersonWriter(): FlatFileItemWriter<Person> {
        // 设置字段提取器
        val fieldExtractor = BeanWrapperFieldExtractor<Person>()
        // 提取这两个字段
        fieldExtractor.setNames(arrayOf("name", "email"))
        fieldExtractor.afterPropertiesSet()

        // 行聚合器
        val lineAggregator = DelimitedLineAggregator<Person>()
        lineAggregator.setDelimiter(";")
        lineAggregator.setFieldExtractor(fieldExtractor)

        return FlatFileItemWriterBuilder<Person>()
            .name("Csv Person 写入器")
            // 设置输出路径
            .resource(FileSystemResource(env.getRequiredProperty(exportCsvPath)))
            // 如果文件已存在, 先删除
            .shouldDeleteIfExists(true)
            // 设置文件标题行(header)
            .headerCallback(StringHeaderWriter(env.getRequiredProperty(exportCsvHeader)))
            // 对应的, 还可以设置 footer
            // .footerCallback(...)
            .lineAggregator(lineAggregator)
            .build()
    }
    // end::读写处理器[]

    // tag::jobstep[]
    /**
     * 每个 step 都包含 reader, processor, writer
     */
    @Bean
    fun csv2MysqlPersonStep1(
//        @Qualifier("csvPersonReader") reader: FlatFileItemReader<Person>,
        // 使用了 triggerReader
        @Qualifier("csvPersonTriggerReader") reader: FlatFileItemReader<Person>,
        @Qualifier("personItemProcessor") processor: ItemProcessor<Person, Person>,
//        @Qualifier("mysqlPersonJdbcWriter") writer: JdbcBatchItemWriter<Person>
        @Qualifier("mysqlPersonJpaWriter") writer: JpaItemWriter<Person>
    ): Step {
        return stepBuilderFactory.get("step1")
            // 一次处理 chunk 条记录
            .chunk<Person, Person>(env.getRequiredProperty(chunck).toInt())
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .build()
    }

    @Bean
    fun csv2CsvPersonStep1(
        @Qualifier("csvPersonTriggerReader") reader: FlatFileItemReader<Person>,
        @Qualifier("personItemProcessor") processor: ItemProcessor<Person, Person>,
        @Qualifier("csvPersonWriter") writer: FlatFileItemWriter<Person>
    ): Step {
        return stepBuilderFactory.get("Csv to Csv Person Step")
            .chunk<Person, Person>(5)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .build()
    }

    @Bean
    fun mysql2CsvPersonStep1(
        reader: JdbcPagingItemReader<Person>,
        @Qualifier("csvPersonWriter") writer: FlatFileItemWriter<Person>
    ): Step {
        return stepBuilderFactory
            .get("Mysql to Csv Person Step")
            .chunk<Person, Person>(env.getRequiredProperty(chunck).toInt())
            .reader(reader)
            // 就不 processor 了
            .writer(writer)
            .build()
    }

    // end::jobstep[]
}

// tag::JobLauncher[]
@Bean
@Throws(Exception::class)
fun jobLauncher(dataSource: DataSource, transactionManager: PlatformTransactionManager): SimpleJobLauncher {
    val jobLauncher = SimpleJobLauncher()
    jobLauncher.setJobRepository(mysqlJobRepository(dataSource, transactionManager))
    // 设置以异步的方式运行
    jobLauncher.setTaskExecutor(SimpleAsyncTaskExecutor())
    jobLauncher.afterPropertiesSet()
    return jobLauncher
}
// end::JobLauncher[]

// tag::JobRepository[]
/**
 *  JobRepository: It provides CRUD operations for JobLauncher, Job, and Step implementations.
 *  当使用了 @EnableBatchProcessing 注解, 就已经自动配置好了一个默认的 JobRepository
 *  我们如果提供数据库, 那就最好了
 */
@Bean
@Throws(Exception::class)
fun mysqlJobRepository(dataSource: DataSource, transactionManager: PlatformTransactionManager): JobRepository? {
    val jobRepositoryFactoryBean = JobRepositoryFactoryBean()
    jobRepositoryFactoryBean.setDataSource(dataSource)
    jobRepositoryFactoryBean.transactionManager = transactionManager
    // 写入到 mysql
    jobRepositoryFactoryBean.setDatabaseType(DatabaseType.MYSQL.toString())
    return jobRepositoryFactoryBean.`object`
}
// end::JobRepository[]
